package com.hlz.flink.chapter05;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @author Hongliang Zhu
 * @create 2023-02-12 20:09
 */
public class TransformFlatMapTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        List<Event> clicks = new ArrayList<>();
        clicks.add(new Event("Mary", "/home/", 1000L));
        clicks.add(new Event("Bob", "/home/", 2000L));
        clicks.add(new Event("Tom", "/cart/", 3000L));

        DataStreamSource<Event> dataStreamSource = env.fromCollection(clicks);

        SingleOutputStreamOperator<String> result = dataStreamSource.flatMap(new MyFlatMapFunc());


        result.print("1");

        env.execute();

    }

    public static class MyFlatMapFunc implements FlatMapFunction<Event, String> {

        @Override
        public void flatMap(Event value, Collector<String> out) throws Exception {
            if (value.getUser().equals("Mary")){
                out.collect(value.getUser());
            }else if (value.getUser().equals("Bob")) {
                out.collect(value.getUser());
                out.collect(value.getUrl());
            }

        }
    }
}
